﻿using MyX.Core;
using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading.Tasks;

namespace MyX.Bus
{
    /// <summary>
    /// 消息队列
    /// </summary>
    public class XMQ : XIMQ
    {
        private static async Task StaticStart()
        {
            if (_run)
                return;

            _run = true;

            var task = Task.Factory.StartNew(
                () =>
                {
                    while (true)
                    {
                        XMQMessage message = _queues.Take();

                        if (message != null)
                        {
                            var handlerMessage = Task.Run(() =>
                            {
                                try
                                {
                                    HaveMessage(message);
                                }
                                catch (Exception)
                                {
                                }
                            });
                        }
                    }
                }, TaskCreationOptions.LongRunning);
            await task;
        }

        private static void HaveMessage(XMQMessage xmqMessage)
        {
            foreach (var bus in _buses)
            {
                if (bus.Topic == xmqMessage.Topic)
                    bus.Receive(xmqMessage.Message);
            }
        }

        public void Send(string topic, XIMessage message)
        {
            _queues.Add(new XMQMessage(topic, message));
        }

        public void Register(XIMessageBus bus)
        {
            _buses.Add(bus);
        }

        public void Start()
        {
            StaticStart();
        }

        public XMQ()
        {

        }

        static XMQ()
        {
            _run = false;
            _queues = new BlockingCollection<XMQMessage>();
            _buses = new List<XIMessageBus>();
        }
        
        private static bool _run;
        private static BlockingCollection<XMQMessage> _queues;
        private static List<XIMessageBus> _buses;
    }

    internal class XMQMessage
    {
        public XMQMessage(string topic, XIMessage message)
        {
            Topic = topic;
            Message = message;
        }

        public string Topic { get; private set; }
        public XIMessage Message { get; private set; }
    }
}
